package com.lncg.service;

import com.lncg.config.AppConfig;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;

import javax.sql.DataSource;
import java.sql.*;
import java.util.ArrayList;
import java.util.List;

/**
 * 数据同步服务
 * 简化版本，包含所有同步逻辑和错误处理
 */
@Service
public class SyncService {
    
    @Autowired
    private AppConfig appConfig;
    
    @Autowired
    @Qualifier("sqlServerDataSource")
    private DataSource sqlServerDataSource;
    
    @Autowired
    @Qualifier("mysqlDataSource")
    private DataSource mysqlDataSource;
    
    @Autowired
    @Qualifier("sqlServer2016DataSource")
    private DataSource sqlServer2016DataSource;
    
    /**
     * 定时同步任务（已移至TaskSchedulerService管理）
     * 保留此方法供TaskSchedulerService调用
     */
    public void scheduledSync() {
        if (!appConfig.isSyncEnabled()) {
            System.out.println("⏸️ 同步已禁用，跳过执行");
            return;
        }
        
        System.out.println("🚀 开始定时同步任务 - " + new java.util.Date());
        
        try {
            syncAll();
            System.out.println("✅ 定时同步任务完成");
        } catch (Exception e) {
            System.err.println("❌ 定时同步任务失败: " + e.getMessage());
            e.printStackTrace();
        }
    }
    
    /**
     * 执行所有同步任务
     */
    public void syncAll() {
        System.out.println("📋 开始执行数据同步...");
        
        try {
            // 验证数据库连接
            validateConnections();
            
            // 执行同步任务1 (SQL Server 2005)
            syncTask1();
            
            // 执行同步任务2 (SQL Server 2005)
            syncTask2();
            
            // 执行同步任务3 (SQL Server 2016)
            syncTask3();
            
            // 执行同步任务4 (SQL Server 2016)
            syncTask4();
            
            System.out.println("🎉 所有同步任务执行完成");
            
        } catch (Exception e) {
            System.err.println("💥 同步任务执行失败: " + e.getMessage());
            e.printStackTrace();
            throw e;
        }
    }
    
    /**
     * 验证数据库连接
     */
    private void validateConnections() {
        System.out.println("🔍 验证数据库连接...");
        
        try (Connection sql2005Conn = sqlServerDataSource.getConnection();
             Connection mysqlConn = mysqlDataSource.getConnection();
             Connection sql2016Conn = sqlServer2016DataSource.getConnection()) {
            
            System.out.println("✅ SQL Server 2005连接正常");
            System.out.println("✅ MySQL连接正常");
            System.out.println("✅ SQL Server 2016连接正常");
            
        } catch (SQLException e) {
            System.err.println("❌ 数据库连接失败: " + e.getMessage());
            System.err.println("💡 请检查:");
            System.err.println("   1. 数据库服务是否启动");
            System.err.println("   2. 网络连接是否正常");
            System.err.println("   3. 用户名密码是否正确");
            System.err.println("   4. 数据库URL是否正确");
            throw new RuntimeException("数据库连接验证失败", e);
        }
    }
    
    /**
     * 同步任务1: 从SQL2005的tbl_cat1_meter_record_current同步到MySQL的wm_user_meter
     * 分批处理，每1000条处理一次，避免内存溢出
     */
    private void syncTask1() {
        System.out.println("📊 开始同步任务1: tbl_cat1_meter_record_current -> wm_user_meter");
        
        Connection sqlConn = null;
        Connection mysqlConn = null;
        PreparedStatement selectStmt = null;
        PreparedStatement checkStmt = null;
        PreparedStatement insertStmt = null;
        PreparedStatement updateStmt = null;
        
        try {
            sqlConn = sqlServerDataSource.getConnection();
            mysqlConn = mysqlDataSource.getConnection();
            
            // 准备MySQL语句 - 使用 ON DUPLICATE KEY UPDATE 实现 upsert (唯一索引是imei)
            String upsertSql = "INSERT INTO wm_user_meter (id, up_key, meter_code, imei, sim, signal_level, online_status, battery_voltage, positive_data, negative_data, online_interval, up_time, dept_id, memo1) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON DUPLICATE KEY UPDATE id = VALUES(id), up_key = VALUES(up_key), meter_code = VALUES(meter_code), sim = VALUES(sim), signal_level = VALUES(signal_level), online_status = VALUES(online_status), battery_voltage = VALUES(battery_voltage), positive_data = VALUES(positive_data), negative_data = VALUES(negative_data), online_interval = VALUES(online_interval), up_time = VALUES(up_time), memo1 = VALUES(memo1)";
            System.out.println("🔧 任务1 - MySQL Upsert SQL: " + upsertSql);
            insertStmt = mysqlConn.prepareStatement(upsertSql);
            
            int totalInsertCount = 0;
            int totalUpdateCount = 0;
            int totalErrorCount = 0;
            int totalProcessed = 0;
            int offset = 0;
            int batchSize = 1000;
            
            // 分批处理数据
            while (true) {
                // 分批查询SQL Server数据 - 使用SQL Server 2005兼容的分页语法
                String selectSql = "SELECT * FROM (SELECT ROW_NUMBER() OVER (ORDER BY imei) AS RowNum, * FROM tbl_cat1_meter_record_current) AS t WHERE RowNum > ? AND RowNum <= ?";
                System.out.println("🔧 任务1 - SQL Server查询SQL: " + selectSql);
                System.out.println("🔧 任务1 - 查询参数: offset=" + offset + ", limit=" + (offset + batchSize));
                selectStmt = sqlConn.prepareStatement(selectSql);
                selectStmt.setInt(1, offset);
                selectStmt.setInt(2, offset + batchSize);
                
                ResultSet rs = selectStmt.executeQuery();
                
                int batchInsertCount = 0;
                int batchUpdateCount = 0;
                int batchErrorCount = 0;
                int batchProcessed = 0;
                
                // 处理当前批次的数据
                while (rs.next()) {
                    batchProcessed++;
                    totalProcessed++;
                    
                    try {
                        // 从SQL Server读取数据
                        Integer id = rs.getInt("id");
                        String upKey = rs.getString("up_key");
                        String meterCode = rs.getString("meter_code");
                        String imei = rs.getString("imei");
                        String sim = rs.getString("sim");
                        String signal = rs.getString("signal"); // 源表字段名
                        String onlineStatus = rs.getString("online_status");
                        String batteryVoltage = rs.getString("battery_voltage");
                        String positiveCumulativeData = rs.getString("positive_cumulative_data"); // 源表字段名
                        String negativeCumulativeData = rs.getString("negative_cumulative_data"); // 源表字段名
                        String onlineInterval = rs.getString("online_interval");
                        Timestamp upTime = rs.getTimestamp("up_time");
                        String memo1 = rs.getString("memo1");
                        
                        // 调试信息：显示前几条记录的详细信息
                        if (totalProcessed <= 3) {
                            System.out.println("🔍 任务1 - 记录 " + totalProcessed + " 详细信息:");
                            System.out.println("   id=" + id + ", imei=" + imei + ", signal=" + signal + ", positive_data=" + positiveCumulativeData + ", negative_data=" + negativeCumulativeData);
                        }
                        
                        if (imei == null || imei.trim().isEmpty()) {
                            System.err.println("⚠️ 跳过空IMEI记录");
                            continue;
                        }
                        
                        // 使用 upsert 操作：如果imei存在则更新，不存在则插入
                        insertStmt.setInt(1, id);
                        insertStmt.setString(2, upKey);
                        insertStmt.setString(3, meterCode);
                        insertStmt.setString(4, imei);
                        insertStmt.setString(5, sim);
                        insertStmt.setString(6, signal); // 映射到 signal_level
                        insertStmt.setString(7, onlineStatus);
                        insertStmt.setString(8, batteryVoltage);
                        insertStmt.setString(9, positiveCumulativeData); // 映射到 positive_data
                        insertStmt.setString(10, negativeCumulativeData); // 映射到 negative_data
                        insertStmt.setString(11, onlineInterval);
                        insertStmt.setTimestamp(12, upTime);
                        insertStmt.setNull(13, Types.VARCHAR); // dept_id 插入时置为 NULL，更新时保持不变
                        insertStmt.setString(14, memo1);
                        
                        int result = insertStmt.executeUpdate();
                        if (result == 1) {
                            // 插入新记录
                            batchInsertCount++;
                            totalInsertCount++;
                        } else if (result == 2) {
                            // 更新现有记录
                            batchUpdateCount++;
                            totalUpdateCount++;
                        }
                    } catch (SQLException e) {
                        batchErrorCount++;
                        totalErrorCount++;
                        System.err.println("❌ 处理记录时出错: " + e.getMessage());
                    }
                }
                
                rs.close();
                selectStmt.close();
                
                // 显示当前批次处理结果
                if (batchProcessed > 0) {
                    System.out.println("📊 批次处理完成 - 处理: " + batchProcessed + " 条，插入: " + batchInsertCount + " 条，更新: " + batchUpdateCount + " 条，错误: " + batchErrorCount + " 条");
                    System.out.println("📈 累计处理: " + totalProcessed + " 条，累计插入: " + totalInsertCount + " 条，累计更新: " + totalUpdateCount + " 条，累计错误: " + totalErrorCount + " 条");
                }
                
                // 如果当前批次处理的数据少于批次大小，说明已经处理完所有数据
                if (batchProcessed < batchSize) {
                    break;
                }
                
                offset += batchSize;
            }
            
            System.out.println("📈 同步任务1完成 - 总计处理: " + totalProcessed + " 条，插入: " + totalInsertCount + " 条，更新: " + totalUpdateCount + " 条，错误: " + totalErrorCount + " 条");
            
        } catch (SQLException e) {
            System.err.println("❌ 同步任务1失败: " + e.getMessage());
            System.err.println("💡 可能的原因:");
            System.err.println("   1. 表名或字段名不正确");
            System.err.println("   2. 数据库权限不足");
            System.err.println("   3. 数据类型不匹配");
            System.err.println("   4. imei字段没有唯一索引");
            throw new RuntimeException("同步任务1执行失败", e);
        } finally {
            closeResources(selectStmt, insertStmt, null, null, sqlConn, mysqlConn);
        }
    }
    
    /**
     * 同步任务2: 根据wm_user_meter中dept_id不为空的记录，从tbl_cat1_meter_record_history同步到wm_user_meter_receive_data
     * 使用wm_sync_status表记录时间戳，按时间增量查询`
     */
    private void syncTask2() {
        System.out.println("📊 开始同步任务2: tbl_cat1_meter_record_history -> wm_user_meter_receive_data");
        
        Connection sqlConn = null;
        Connection mysqlConn = null;
        PreparedStatement selectUserStmt = null;
        PreparedStatement selectHistoryStmt = null;
        PreparedStatement insertStmt = null;
        PreparedStatement selectSyncTimeStmt = null;
        PreparedStatement updateSyncTimeStmt = null;
        
        try {
            sqlConn = sqlServerDataSource.getConnection();
            mysqlConn = mysqlDataSource.getConnection();
            
            // 获取上次同步时间
            String selectSyncTimeSql = "SELECT last_sync_time FROM wm_sync_status WHERE table_name = 'wm_user_meter_receive_data'";
            System.out.println("🔧 任务2 - 查询同步时间SQL: " + selectSyncTimeSql);
            selectSyncTimeStmt = mysqlConn.prepareStatement(selectSyncTimeSql);
            ResultSet syncTimeRs = selectSyncTimeStmt.executeQuery();
            
            Timestamp lastSyncTime = null;
            if (syncTimeRs.next()) {
                lastSyncTime = syncTimeRs.getTimestamp("last_sync_time");
            }
            syncTimeRs.close();
            selectSyncTimeStmt.close();
            
            if (lastSyncTime == null) {
                // 如果没有记录，从当前时间开始
                lastSyncTime = new Timestamp(System.currentTimeMillis());
                System.out.println("📅 首次同步，从当前时间开始: " + lastSyncTime);
            } else {
                System.out.println("📅 上次同步时间: " + lastSyncTime);
            }
            
            // 查询wm_user_meter表中dept_id不为空的记录总数
            String countUserSql = "SELECT COUNT(*) as total FROM wm_user_meter WHERE dept_id IS NOT NULL AND dept_id != ''";
            System.out.println("🔧 任务2 - 统计用户记录SQL: " + countUserSql);
            PreparedStatement countStmt = mysqlConn.prepareStatement(countUserSql);
            ResultSet countRs = countStmt.executeQuery();
            int totalUserCount = 0;
            if (countRs.next()) {
                totalUserCount = countRs.getInt("total");
            }
            countRs.close();
            countStmt.close();
            System.out.println("📊 任务2 - wm_user_meter表中dept_id不为空的记录总数: " + totalUserCount);
            
            if (totalUserCount == 0) {
                System.out.println("⚠️ 任务2 - 没有找到dept_id不为空的用户记录，跳过同步");
                // 关闭已创建的连接，避免连接泄漏
                closeResources(null, null, null, null, sqlConn, mysqlConn);
                return;
            }
            
            // 准备插入语句 - 根据实际目标表结构
            String insertSql = "INSERT INTO wm_user_meter_receive_data (row_id, up_key, meter_code, imei, sim, signal_level, online_status, battery_voltage, positive_data, negative_data, online_interval, crc, raw_data, up_time, memo1, memo2, memo3, memo4, memo5, create_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW())";
            System.out.println("🔧 任务2 - MySQL插入SQL: " + insertSql);
            insertStmt = mysqlConn.prepareStatement(insertSql);
            
            int totalInsertCount = 0;
            int totalErrorCount = 0;
            int totalProcessed = 0;
            int userOffset = 0;
            int userBatchSize = 100; // 每次处理100个用户ID
            
            System.out.println("🔧 任务2 - 初始化: userOffset=" + userOffset + ", userBatchSize=" + userBatchSize);
            System.out.println("🔧 任务2 - 强制设置 userOffset = 0");
            userOffset = 0;
            
            // 分批查询用户ID
            while (true) {
                // 分批查询MySQL中dept_id不为空的记录 - 使用LIMIT offset, count语法
                String selectUserSql = "SELECT id FROM wm_user_meter WHERE dept_id IS NOT NULL AND dept_id != '' ORDER BY id LIMIT ?, ?";
                System.out.println("🔧 任务2 - 查询用户ID SQL: " + selectUserSql);
                System.out.println("🔧 任务2 - 用户查询参数: offset=" + userOffset + ", batchSize=" + userBatchSize);
                selectUserStmt = mysqlConn.prepareStatement(selectUserSql);
                selectUserStmt.setInt(1, userOffset);
                selectUserStmt.setInt(2, userBatchSize);
                
                ResultSet userRs = selectUserStmt.executeQuery();
                
                List<Integer> userIds = new ArrayList<>();
                int userCount = 0;
                while (userRs.next()) {
                    int userId = userRs.getInt("id");
                    userIds.add(userId);
                    userCount++;
                    
                    // 显示前10个用户的详细信息
                    if (userCount <= 10) {
                        System.out.println("🔍 任务2 - 用户记录 " + userCount + ": id=" + userId);
                    }
                }
                userRs.close();
                selectUserStmt.close();
                
                if (userIds.isEmpty()) {
                    System.out.println("📋 任务2 - 没有找到dept_id不为空的用户记录");
                    break; // 没有更多用户ID了
                }
                
                // 如果返回的记录数少于批次大小，说明已经查询完了所有数据
                if (userIds.size() < userBatchSize) {
                    System.out.println("📋 任务2 - 已查询完所有用户记录，当前批次返回 " + userIds.size() + " 条记录");
                }
                
                System.out.println("👥 处理用户ID批次: " + userIds.size() + " 个用户");
                if (userCount > 10) {
                    System.out.println("📊 任务2 - 总共找到 " + userCount + " 个dept_id不为空的用户记录（仅显示前10个）");
                }
                
                // 为每个用户ID查询历史记录（也使用分页）
                for (Integer userId : userIds) {
                    int historyOffset = 0;
                    int historyBatchSize = 1000; // 每个用户的历史记录也分页查询
                    int userInsertCount = 0; // 移到外层循环，记录该用户的总插入数
                    
                    while (true) {
                        try {
                            // 分页查询历史记录 - 使用SQL Server 2005兼容的分页语法，按时间增量查询（无排序）
                            String selectHistorySql = "SELECT * FROM (SELECT ROW_NUMBER() OVER (ORDER BY (SELECT NULL)) AS RowNum, * FROM tbl_cat1_meter_record_history WHERE row_id = ? AND up_time > ?) AS t WHERE RowNum > ? AND RowNum <= ?";
                            System.out.println("🔧 任务2 - 查询历史记录SQL: " + selectHistorySql);
                            System.out.println("🔧 任务2 - 历史记录查询参数: userId=" + userId + ", lastSyncTime=" + lastSyncTime + ", offset=" + historyOffset + ", limit=" + (historyOffset + historyBatchSize));
                            selectHistoryStmt = sqlConn.prepareStatement(selectHistorySql);
                            selectHistoryStmt.setInt(1, userId);
                            selectHistoryStmt.setTimestamp(2, lastSyncTime);
                            selectHistoryStmt.setInt(3, historyOffset);
                            selectHistoryStmt.setInt(4, historyOffset + historyBatchSize);
                            
                            ResultSet historyRs = selectHistoryStmt.executeQuery();
                            
                            int batchProcessed = 0;
                            
                            while (historyRs.next()) {
                                try {
                                    // 根据实际SQL Server历史表结构映射字段
                                    String upKey = historyRs.getString("up_key");
                                    String meterCode = historyRs.getString("meter_code");
                                    String imei = historyRs.getString("imei");
                                    String sim = historyRs.getString("sim");
                                    Integer signal = historyRs.getInt("signal");
                                    Integer onlineStatus = historyRs.getInt("online_status");
                                    Double batteryVoltage = historyRs.getDouble("battery_voltage");
                                    Double positiveData = historyRs.getDouble("positive_cumulative_data");
                                    Double negativeData = historyRs.getDouble("negative_cumulative_data");
                                    Integer onlineInterval = historyRs.getInt("online_interval");
                                    String crc = historyRs.getString("crc");
                                    String rawData = historyRs.getString("raw_data");
                                    Timestamp upTime = historyRs.getTimestamp("up_time");
                                    String memo1 = historyRs.getString("memo1");
                                    String memo2 = historyRs.getString("memo2");
                                    String memo3 = historyRs.getString("memo3");
                                    String memo4 = historyRs.getString("memo4");
                                    String memo5 = historyRs.getString("memo5");
                                    
                                    insertStmt.setInt(1, userId);           // row_id
                                    insertStmt.setString(2, upKey);         // up_key
                                    insertStmt.setString(3, meterCode);     // meter_code
                                    insertStmt.setString(4, imei);          // imei
                                    insertStmt.setString(5, sim);           // sim
                                    insertStmt.setInt(6, signal);           // signal_level
                                    insertStmt.setInt(7, onlineStatus);     // online_status
                                    insertStmt.setDouble(8, batteryVoltage); // battery_voltage
                                    insertStmt.setDouble(9, positiveData);  // positive_data
                                    insertStmt.setDouble(10, negativeData); // negative_data
                                    insertStmt.setInt(11, onlineInterval);  // online_interval
                                    insertStmt.setString(12, crc);          // crc
                                    insertStmt.setString(13, rawData);      // raw_data
                                    insertStmt.setTimestamp(14, upTime);    // up_time
                                    insertStmt.setString(15, memo1);        // memo1
                                    insertStmt.setString(16, memo2);        // memo2
                                    insertStmt.setString(17, memo3);        // memo3
                                    insertStmt.setString(18, memo4);        // memo4
                                    insertStmt.setString(19, memo5);        // memo5
                                    insertStmt.executeUpdate();
                                    
                                    userInsertCount++;
                                    totalInsertCount++;
                                    totalProcessed++;
                                    batchProcessed++;
                                    
                                    // 每1000条数据给一个提示
                                    if (totalProcessed % 1000 == 0) {
                                        System.out.println("📊 已处理 " + totalProcessed + " 条历史记录 - 插入: " + totalInsertCount + " 条，错误: " + totalErrorCount + " 条");
                                    }
                                } catch (SQLException e) {
                                    totalErrorCount++;
                                    System.err.println("❌ 插入历史记录时出错: " + e.getMessage());
                                }
                            }
                            
                            historyRs.close();
                            selectHistoryStmt.close();
                            
                            // 如果当前批次处理的数据少于批次大小，说明该用户的历史记录已经处理完
                            if (batchProcessed < historyBatchSize) {
                                break;
                            }
                            
                            historyOffset += historyBatchSize;
                            
                        } catch (SQLException e) {
                            totalErrorCount++;
                            System.err.println("❌ 查询用户 " + userId + " 历史记录时出错: " + e.getMessage());
                            break; // 跳过这个用户，继续处理下一个
                        }
                    }
                    
                    if (userInsertCount > 0) {
                        System.out.println("👤 用户ID " + userId + " 同步了 " + userInsertCount + " 条历史记录");
                    }
                }
                
                // 只有当查询到记录时才更新偏移量
                if (!userIds.isEmpty()) {
                    userOffset += userBatchSize;
                } else {
                    // 如果没有查询到记录，退出循环
                    break;
                }
            }
            
            System.out.println("📈 同步任务2完成 - 总共插入: " + totalInsertCount + " 条，错误: " + totalErrorCount + " 条");
            
            // 更新同步时间
            Timestamp currentTime = new Timestamp(System.currentTimeMillis());
            String updateSyncTimeSql = "INSERT INTO wm_sync_status (table_name, last_sync_time, last_sync_value, created_time, updated_time) VALUES ('wm_user_meter_receive_data', ?, ?, NOW(), NOW()) ON DUPLICATE KEY UPDATE last_sync_time = VALUES(last_sync_time), last_sync_value = VALUES(last_sync_value), updated_time = NOW()";
            System.out.println("🔧 任务2 - 更新同步时间SQL: " + updateSyncTimeSql);
            System.out.println("🔧 任务2 - 更新参数: currentTime=" + currentTime + ", totalInsertCount=" + totalInsertCount);
            updateSyncTimeStmt = mysqlConn.prepareStatement(updateSyncTimeSql);
            updateSyncTimeStmt.setTimestamp(1, currentTime);
            updateSyncTimeStmt.setString(2, String.valueOf(totalInsertCount)); // 使用插入的记录数作为同步值
            updateSyncTimeStmt.executeUpdate();
            updateSyncTimeStmt.close();
            System.out.println("📅 已更新同步时间: " + currentTime + "，同步记录数: " + totalInsertCount);
            
        } catch (SQLException e) {
            System.err.println("❌ 同步任务2失败: " + e.getMessage());
            System.err.println("💡 可能的原因:");
            System.err.println("   1. 表名或字段名不正确");
            System.err.println("   2. 数据库权限不足");
            System.err.println("   3. 数据类型不匹配");
            System.err.println("   4. SQL Server版本不支持OFFSET/FETCH语法");
            throw new RuntimeException("同步任务2执行失败", e);
        } finally {
            closeResources(selectUserStmt, selectHistoryStmt, insertStmt, updateSyncTimeStmt, sqlConn, mysqlConn);
        }
    }
    
    /**
     * 关闭数据库资源
     */
    private void closeResources(PreparedStatement... statements) {
        for (PreparedStatement stmt : statements) {
            if (stmt != null) {
                try {
                    stmt.close();
                } catch (SQLException e) {
                    System.err.println("⚠️ 关闭PreparedStatement时出错: " + e.getMessage());
                }
            }
        }
    }
    
    /**
     * 关闭数据库资源
     */
    private void closeResources(PreparedStatement stmt1, PreparedStatement stmt2, PreparedStatement stmt3, PreparedStatement stmt4, Connection conn1, Connection conn2) {
        closeResources(stmt1, stmt2, stmt3, stmt4);
        if (conn1 != null) {
            try {
                conn1.close();
            } catch (SQLException e) {
                System.err.println("⚠️ 关闭SQL Server连接时出错: " + e.getMessage());
            }
        }
        if (conn2 != null) {
            try {
                conn2.close();
            } catch (SQLException e) {
                System.err.println("⚠️ 关闭MySQL连接时出错: " + e.getMessage());
            }
        }
    }
    
    /**
     * 同步任务3: MulityRoute -> wm_mulity_route (UPSERT同步)
     */
    private void syncTask3() {
        System.out.println("📊 开始同步任务3: MulityRoute -> wm_mulity_route");
        
        Connection sql2016Conn = null;
        Connection mysqlConn = null;
        PreparedStatement selectStmt = null;
        PreparedStatement insertStmt = null;
        
        try {
            sql2016Conn = sqlServer2016DataSource.getConnection();
            mysqlConn = mysqlDataSource.getConnection();
            
            // 准备MySQL语句 - 以 IMEI 为唯一键做 UPSERT，不插入 id 列
            String upsertSql = "INSERT INTO wm_mulity_route (systime, sim, imei, uptype, version, signel_level, battery, sequence, receive, dept_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON DUPLICATE KEY UPDATE systime = VALUES(systime), sim = VALUES(sim), uptype = VALUES(uptype), version = VALUES(version), signel_level = VALUES(signel_level), battery = VALUES(battery), sequence = VALUES(sequence), receive = VALUES(receive)";
            System.out.println("🔧 任务3 - MySQL Upsert SQL: " + upsertSql);
            insertStmt = mysqlConn.prepareStatement(upsertSql);
            
            int totalInsertCount = 0;
            int totalUpdateCount = 0;
            int totalErrorCount = 0;
            int totalProcessed = 0;
            int offset = 0;
            int batchSize = 1000;
            
            // 分批处理数据
            while (true) {
                // 分批查询SQL Server 2016数据
                String selectSql = "SELECT * FROM (SELECT ROW_NUMBER() OVER (ORDER BY IMEI) AS RowNum, * FROM MulityRoute) AS t WHERE RowNum > ? AND RowNum <= ?";
                System.out.println("🔧 任务3 - SQL Server 2016查询SQL: " + selectSql);
                System.out.println("🔧 任务3 - 查询参数: offset=" + offset + ", limit=" + (offset + batchSize));
                selectStmt = sql2016Conn.prepareStatement(selectSql);
                selectStmt.setInt(1, offset);
                selectStmt.setInt(2, offset + batchSize);
                
                ResultSet rs = selectStmt.executeQuery();
                
                
                int batchInsertCount = 0;
                int batchUpdateCount = 0;
                int batchErrorCount = 0;
                int batchProcessed = 0;
                
                // 处理当前批次的数据
                while (rs.next()) {
                    batchProcessed++;
                    totalProcessed++;
                    
                    try {
                        // 从SQL Server 2016读取数据 - 根据实际字段结构
                        Timestamp systime = rs.getTimestamp("systime");
                        String sim = rs.getString("sim");
                        String imei = rs.getString("imei");
                        String uptype = rs.getString("uptype");
                        String version = rs.getString("version");
                        String signel = rs.getString("signel"); // 源表字段名，映射到signel_level
                        String battery = rs.getString("battery");
                        String sequence = rs.getString("sequence");
                        String receive = rs.getString("receive");
                        
                        
                        if (imei == null || imei.trim().isEmpty()) {
                            System.err.println("⚠️ 跳过空IMEI记录");
                            continue;
                        }
                        
                        // 使用 upsert 操作：如果IMEI存在则更新，不存在则插入
                        insertStmt.setTimestamp(1, systime);
                        insertStmt.setString(2, sim);
                        insertStmt.setString(3, imei);
                        insertStmt.setString(4, uptype);
                        insertStmt.setString(5, version);
                        insertStmt.setString(6, signel); // 映射到 signel_level
                        insertStmt.setString(7, battery);
                        insertStmt.setString(8, sequence);
                        insertStmt.setString(9, receive);
                        insertStmt.setNull(10, Types.VARCHAR); // dept_id 插入时置为 NULL
                        
                        int result = insertStmt.executeUpdate();
                        if (result == 1) {
                            // 插入新记录
                            batchInsertCount++;
                            totalInsertCount++;
                        } else if (result == 2) {
                            // 更新现有记录
                            batchUpdateCount++;
                            totalUpdateCount++;
                        }
                    } catch (SQLException e) {
                        batchErrorCount++;
                        totalErrorCount++;
                        System.err.println("❌ 处理记录时出错: " + e.getMessage());
                    }
                }
                
                rs.close();
                selectStmt.close();
                
                // 显示当前批次处理结果
                if (batchProcessed > 0) {
                    System.out.println("📊 任务3 - 批次处理完成 - 处理: " + batchProcessed + " 条，插入: " + batchInsertCount + " 条，更新: " + batchUpdateCount + " 条，错误: " + batchErrorCount + " 条");
                    System.out.println("📈 任务3 - 累计处理: " + totalProcessed + " 条，累计插入: " + totalInsertCount + " 条，累计更新: " + totalUpdateCount + " 条，累计错误: " + totalErrorCount + " 条");
                }
                
                // 如果当前批次处理的数据少于批次大小，说明已经处理完所有数据
                if (batchProcessed < batchSize) {
                    break;
                }
                
                offset += batchSize;
            }
            
            System.out.println("📈 同步任务3完成 - 总共处理: " + totalProcessed + " 条，插入: " + totalInsertCount + " 条，更新: " + totalUpdateCount + " 条，错误: " + totalErrorCount + " 条");
            
        } catch (Exception e) {
            System.err.println("❌ 同步任务3失败: " + e.getMessage());
            System.err.println("💡 可能的原因:");
            System.err.println("   1. 表名或字段名不正确");
            System.err.println("   2. 数据库权限不足");
            System.err.println("   3. 数据类型不匹配");
            System.err.println("   4. IMEI字段没有唯一索引");
            e.printStackTrace();
            throw new RuntimeException("同步任务3执行失败", e);
        } finally {
            closeResources(selectStmt, insertStmt, null, null, sql2016Conn, mysqlConn);
        }
    }
    
    /**
     * 同步任务4: MulityRoute_His -> wm_mulity_route_his (增量同步)
     * 根据wm_mulity_route表dept_id不为空的imei，从MulityRoute_His表按时间增量同步数据
     */
    private void syncTask4() {
        System.out.println("📊 开始同步任务4: MulityRoute_His -> wm_mulity_route_his");
        System.out.println("🔧 任务4 - 正在连接数据库...");
        
        Connection sql2016Conn = null;
        Connection mysqlConn = null;
        PreparedStatement selectImeiStmt = null;
        PreparedStatement selectHistoryStmt = null;
        PreparedStatement insertStmt = null;
        PreparedStatement selectSyncTimeStmt = null;
        PreparedStatement updateSyncTimeStmt = null;
        PreparedStatement insertMeterDataStmt = null;
        PreparedStatement selectMeterDataStmt = null;
        
        try {
            sql2016Conn = sqlServer2016DataSource.getConnection();
            mysqlConn = mysqlDataSource.getConnection();
            
            // 1. 获取上次同步时间
            String selectSyncTimeSql = "SELECT last_sync_time FROM wm_sync_status WHERE table_name = 'wm_mulity_route_his'";
            System.out.println("🔧 任务4 - 查询同步时间SQL: " + selectSyncTimeSql);
            selectSyncTimeStmt = mysqlConn.prepareStatement(selectSyncTimeSql);
            ResultSet syncTimeRs = selectSyncTimeStmt.executeQuery();
            
            Timestamp lastSyncTime = null;
            if (syncTimeRs.next()) {
                lastSyncTime = syncTimeRs.getTimestamp("last_sync_time");
            }
            syncTimeRs.close();
            selectSyncTimeStmt.close();
            
            if (lastSyncTime == null) {
                // 如果没有记录，从目标表当前最大时间戳开始
                String maxTimeSql = "SELECT MAX(systime) FROM wm_mulity_route_his";
                System.out.println("🔧 任务4 - 查询目标表最大时间SQL: " + maxTimeSql);
                PreparedStatement maxTimeStmt = mysqlConn.prepareStatement(maxTimeSql);
                ResultSet maxTimeRs = maxTimeStmt.executeQuery();
                if (maxTimeRs.next()) {
                    lastSyncTime = maxTimeRs.getTimestamp(1);
                }
                maxTimeRs.close();
                maxTimeStmt.close();
                
                if (lastSyncTime == null) {
                    System.out.println("📅 首次同步，从当前时间开始");
                    lastSyncTime = new Timestamp(System.currentTimeMillis());
                } else {
                    System.out.println("📅 从目标表最大时间开始: " + lastSyncTime);
                }
            } else {
                System.out.println("📅 上次同步时间: " + lastSyncTime);
            }
            
            // 2. 获取wm_user_meter表中的IMEI列表
            String selectImeiSql = "SELECT DISTINCT imei FROM wm_mulity_route WHERE dept_id IS NOT NULL AND dept_id != ''";
            System.out.println("🔧 任务4 - 查询IMEI列表SQL: " + selectImeiSql);
            selectImeiStmt = mysqlConn.prepareStatement(selectImeiSql);
            ResultSet imeiRs = selectImeiStmt.executeQuery();
            
            List<String> imeiList = new ArrayList<>();
            while (imeiRs.next()) {
                String imei = imeiRs.getString("imei");
                if (imei != null && !imei.trim().isEmpty()) {
                    imeiList.add(imei.trim());
                }
            }
            imeiRs.close();
            selectImeiStmt.close();
            
            System.out.println("📊 任务4 - 找到 " + imeiList.size() + " 个有效的IMEI");
            
            if (imeiList.isEmpty()) {
                System.out.println("⚠️ 任务4 - 没有找到有效的IMEI，跳过同步");
                return;
            }
            
            // 3. 准备插入语句
            String insertSql = "INSERT IGNORE INTO wm_mulity_route_his (id, systime, sim, imei, uptype, version, signel_level, battery, sequence, receive, dept_id, memo1, memo2, memo3, memo4) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
            System.out.println("🔧 任务4 - MySQL插入SQL: " + insertSql);
            insertStmt = mysqlConn.prepareStatement(insertSql);
            
            // 3.1 准备表数据插入语句
            String insertMeterSql = "INSERT IGNORE INTO wm_mulity_route_meter_data (id, hisid, metercode, routeid, sequence, data, status, mulit, typecode, strong, remark) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
            insertMeterDataStmt = mysqlConn.prepareStatement(insertMeterSql);
            
            int totalInsertCount = 0;
            int totalUpdateCount = 0; // 历史表保留统计名以兼容日志
            int totalErrorCount = 0;
            int totalProcessed = 0;
            int totalMeterInserted = 0;
            Timestamp maxSystime = null; // 记录本次同步的最大时间戳
            
            // 4. 分批处理IMEI列表
            int offset = 0;
            int batchSize = 1000;
            
            while (true) {
                // 构建查询SQL - 使用ROW_NUMBER()分页
                StringBuilder selectHistorySql = new StringBuilder();
                selectHistorySql.append("SELECT * FROM (SELECT ROW_NUMBER() OVER (ORDER BY [systime]) AS RowNum, [id], [systime], [sim], [imei], [uptype], [version], [signel], [battery], [sequence], [receive] FROM [NBIoTDB].[dbo].[MulityRoute_His] WHERE [imei] IN (");
                for (int j = 0; j < imeiList.size(); j++) {
                    if (j > 0) selectHistorySql.append(", ");
                    selectHistorySql.append("?");
                }
                selectHistorySql.append(")");
                if (lastSyncTime != null) {
                    selectHistorySql.append(" AND [systime] > ?");
                }
                selectHistorySql.append(") AS t WHERE RowNum > ? AND RowNum <= ?");
                
                System.out.println("🔧 任务4 - SQL Server 2016查询SQL: " + selectHistorySql.toString());
                System.out.println("🔧 任务4 - 查询参数: offset=" + offset + ", limit=" + (offset + batchSize) + ", lastSyncTime=" + lastSyncTime);
                selectHistoryStmt = sql2016Conn.prepareStatement(selectHistorySql.toString());
                
                // 设置IMEI参数
                for (int j = 0; j < imeiList.size(); j++) {
                    selectHistoryStmt.setString(j + 1, imeiList.get(j));
                }
                
                // 设置时间参数
                int paramIndex = imeiList.size() + 1;
                if (lastSyncTime != null) {
                    selectHistoryStmt.setTimestamp(paramIndex, lastSyncTime);
                    paramIndex++;
                }
                
                // 设置分页参数
                selectHistoryStmt.setInt(paramIndex, offset);
                selectHistoryStmt.setInt(paramIndex + 1, offset + batchSize);
                
                ResultSet historyRs = selectHistoryStmt.executeQuery();
                
                int batchInsertCount = 0;
                int batchUpdateCount = 0;
                int batchErrorCount = 0;
                int batchProcessed = 0;
                int batchMeterInserted = 0;
                java.util.List<Integer> hisIdList = new java.util.ArrayList<>();
                
                // 处理查询结果
                while (historyRs.next()) {
                    batchProcessed++;
                    totalProcessed++;
                    
                    try {
                        // 从SQL Server 2016读取数据
                        Integer id = historyRs.getInt("id");
                        Timestamp systime = historyRs.getTimestamp("systime");
                        String sim = historyRs.getString("sim");
                        String imei = historyRs.getString("imei");
                        String uptype = historyRs.getString("uptype");
                        String version = historyRs.getString("version");
                        String signel = historyRs.getString("signel"); // 源表字段名，映射到signel_level
                        String battery = historyRs.getString("battery");
                        String sequence = historyRs.getString("sequence");
                        String receive = historyRs.getString("receive");
                        
                        // 记录最大systime
                        if (systime != null) {
                            if (maxSystime == null || systime.after(maxSystime)) {
                                maxSystime = systime;
                            }
                        }
                        
                        if (imei == null || imei.trim().isEmpty()) {
                            System.err.println("⚠️ 跳过空IMEI记录");
                            continue;
                        }
                        
                        // 插入历史表
                        insertStmt.setInt(1, id);
                        insertStmt.setTimestamp(2, systime);
                        insertStmt.setString(3, sim);
                        insertStmt.setString(4, imei);
                        insertStmt.setString(5, uptype);
                        insertStmt.setString(6, version);
                        insertStmt.setString(7, signel); // 映射到 signel_level
                        insertStmt.setString(8, battery);
                        insertStmt.setString(9, sequence);
                        insertStmt.setString(10, receive);
                        insertStmt.setNull(11, Types.VARCHAR); // dept_id 插入时置为 NULL
                        insertStmt.setNull(12, Types.VARCHAR); // memo1 插入时置为 NULL
                        insertStmt.setNull(13, Types.VARCHAR); // memo2 插入时置为 NULL
                        insertStmt.setNull(14, Types.VARCHAR); // memo3 插入时置为 NULL
                        insertStmt.setNull(15, Types.VARCHAR); // memo4 插入时置为 NULL
                        
                        int result = insertStmt.executeUpdate();
                        if (result >= 1) {
                            // 插入新记录
                            batchInsertCount++;
                            totalInsertCount++;
                        }
                        // 收集 hisid，稍后去源表查表数据
                        if (id != null) {
                            hisIdList.add(id);
                        }
                    } catch (SQLException e) {
                        batchErrorCount++;
                        totalErrorCount++;
                        System.err.println("❌ 处理记录时出错: " + e.getMessage());
                    }
                }
                
                historyRs.close();
                selectHistoryStmt.close();
                
                // 使用收集的 hisid 去 SQL Server 2016 查询表数据并插入到 wm_mulity_route_meter_data
                if (!hisIdList.isEmpty()) {
                    int inBatchSize = 500;
                    for (int start = 0; start < hisIdList.size(); start += inBatchSize) {
                        int end = Math.min(start + inBatchSize, hisIdList.size());
                        java.util.List<Integer> sub = hisIdList.subList(start, end);
                        StringBuilder meterDataSql = new StringBuilder();
                        meterDataSql.append("SELECT [id],[hisid],[metercode],[routeid],[sequence],[data],[status],[mulit],[typecode],[strong] FROM [NBIoTDB].[dbo].[MulityRoutMeterData] WHERE [hisid] IN (");
                        for (int i = 0; i < sub.size(); i++) {
                            if (i > 0) meterDataSql.append(",");
                            meterDataSql.append("?");
                        }
                        meterDataSql.append(")");
                        
                        selectMeterDataStmt = sql2016Conn.prepareStatement(meterDataSql.toString());
                        for (int i = 0; i < sub.size(); i++) {
                            selectMeterDataStmt.setInt(i + 1, sub.get(i));
                        }
                        ResultSet mrs = selectMeterDataStmt.executeQuery();
                        while (mrs.next()) {
                            try {
                                int mid = mrs.getInt("id");
                                int mhisid = mrs.getInt("hisid");
                                String metercode = mrs.getString("metercode");
                                Integer routeid = (Integer) mrs.getObject("routeid");
                                Integer sequence = (Integer) mrs.getObject("sequence");
                                java.math.BigDecimal data = mrs.getBigDecimal("data");
                                Integer status = (Integer) mrs.getObject("status");
                                Integer mulit = (Integer) mrs.getObject("mulit");
                                Integer typecode = (Integer) mrs.getObject("typecode");
                                Integer strong = (Integer) mrs.getObject("strong");
                                
                                insertMeterDataStmt.setInt(1, mid);
                                insertMeterDataStmt.setInt(2, mhisid);
                                insertMeterDataStmt.setString(3, metercode);
                                if (routeid == null) insertMeterDataStmt.setNull(4, Types.INTEGER); else insertMeterDataStmt.setInt(4, routeid);
                                if (sequence == null) insertMeterDataStmt.setNull(5, Types.INTEGER); else insertMeterDataStmt.setInt(5, sequence);
                                insertMeterDataStmt.setBigDecimal(6, data);
                                if (status == null) insertMeterDataStmt.setNull(7, Types.INTEGER); else insertMeterDataStmt.setInt(7, status);
                                if (mulit == null) insertMeterDataStmt.setNull(8, Types.INTEGER); else insertMeterDataStmt.setInt(8, mulit);
                                if (typecode == null) insertMeterDataStmt.setNull(9, Types.INTEGER); else insertMeterDataStmt.setInt(9, typecode);
                                if (strong == null) insertMeterDataStmt.setNull(10, Types.INTEGER); else insertMeterDataStmt.setInt(10, strong);
                                insertMeterDataStmt.setNull(11, Types.VARCHAR);
                                
                                int ir = insertMeterDataStmt.executeUpdate();
                                if (ir > 0) {
                                    batchMeterInserted++;
                                    totalMeterInserted++;
                                }
                            } catch (SQLException ee) {
                                totalErrorCount++;
                                System.err.println("❌ 任务4 - 插入表数据失败: " + ee.getMessage());
                            }
                        }
                        mrs.close();
                        selectMeterDataStmt.close();
                    }
                }

                // 显示当前批次处理结果
                if (batchProcessed > 0) {
                    System.out.println("📊 任务4 - 批次处理完成 - 历史处理: " + batchProcessed + " 条，历史插入: " + batchInsertCount + " 条，错误: " + batchErrorCount + " 条；表数据插入: " + batchMeterInserted + " 条");
                    System.out.println("📈 任务4 - 累计历史处理: " + totalProcessed + " 条，累计历史插入: " + totalInsertCount + " 条，累计表数据插入: " + totalMeterInserted + " 条，累计错误: " + totalErrorCount + " 条");
                }
                
                // 如果当前批次处理的数据少于批次大小，说明已经处理完所有数据
                if (batchProcessed < batchSize) {
                    break;
                }
                
                offset += batchSize;
            }
            
            System.out.println("📈 同步任务4完成 - 历史处理: " + totalProcessed + " 条，历史插入: " + totalInsertCount + " 条，表数据插入: " + totalMeterInserted + " 条，错误: " + totalErrorCount + " 条");
            
            // 5. 更新同步时间
            if (maxSystime != null) {
                String updateSyncTimeSql = "INSERT INTO wm_sync_status (table_name, last_sync_time, last_sync_value, created_time, updated_time) VALUES ('wm_mulity_route_his', ?, ?, NOW(), NOW()) ON DUPLICATE KEY UPDATE last_sync_time = VALUES(last_sync_time), last_sync_value = VALUES(last_sync_value), updated_time = NOW()";
                System.out.println("🔧 任务4 - 更新同步时间SQL: " + updateSyncTimeSql);
                System.out.println("🔧 任务4 - 更新参数: maxSystime=" + maxSystime + ", totalProcessed=" + totalProcessed);
                updateSyncTimeStmt = mysqlConn.prepareStatement(updateSyncTimeSql);
                updateSyncTimeStmt.setTimestamp(1, maxSystime);
                updateSyncTimeStmt.setString(2, String.valueOf(totalProcessed)); // 使用处理的记录数作为同步值
                updateSyncTimeStmt.executeUpdate();
                updateSyncTimeStmt.close();
                System.out.println("📅 已更新同步时间: " + maxSystime + "，同步记录数: " + totalProcessed);
            }
            
        } catch (SQLException e) {
            System.err.println("❌ 同步任务4失败: " + e.getMessage());
            System.err.println("💡 可能的原因:");
            System.err.println("   1. 表名或字段名不正确");
            System.err.println("   2. 数据库权限不足");
            System.err.println("   3. 数据类型不匹配");
            System.err.println("   4. id字段没有唯一索引");
            throw new RuntimeException("同步任务4执行失败", e);
        } finally {
            closeResources(selectImeiStmt, selectHistoryStmt, insertStmt, updateSyncTimeStmt, sql2016Conn, mysqlConn);
            closeResources(insertMeterDataStmt, selectMeterDataStmt);
        }
    }
}
